
package clientv3

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	"go.etcd.io/etcd/client/v3/credentials"
	"go.etcd.io/etcd/client/v3/internal/endpoint"
	"go.etcd.io/etcd/client/v3/internal/resolver"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	grpccredentials "google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"google.golang.org/grpc/status"
)

var (
	ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
	ErrOldCluster           = errors.New("etcdclient: old cluster version")
)

// 客户端提供并管理etcd v3客户端会话。
type Client struct {
	// 负责完成集群节点的管理操作，例如 ， 增删改查集群中的节点信息。
	Cluster
	// 负责处理键值对操作，例如，增删改查键值对数据、触发压缩操作等
	KV
	// 负责租约相关的操作，例如，新建和回收租约、为租约续期等。
	Lease
	// 负责为指定的 Key 添加 Watcher
	Watcher
	// 负责权限相关的操作，例如，添加用户或角色、为用户分配角色等 。
	Auth
	// 提供了获取 Alarm 列表、关闭 Alarm 的 功能，该接口还提供了
	// 读取指定节点状态的功 能、 触发指定节点进行压缩的功能，以及读取指定节点快照
	// 的功能。
	Maintenance
	// 与服务端交互的连接，
	conn *grpc.ClientConn
	// Client 的相关配置都封装在该字段中
	cfg      Config
	creds    grpccredentials.TransportCredentials
	resolver *resolver.EtcdManualResolver

	epMu      *sync.RWMutex
	endpoints []string

	ctx    context.Context
	cancel context.CancelFunc

	// 当前客户端的用户名和密码， 当客户端正式
	// 发起请求之前，会先通过用户名和密码检测相应权限。
	Username string
	Password        string

	authTokenBundle credentials.Bundle

	callOpts []grpc.CallOption

	lgMu *sync.RWMutex
	lg   *zap.Logger
}

// New根据给定配置创建一个新的etcdv3客户端。
func New(cfg Config) (*Client, error) {
	if len(cfg.Endpoints) == 0 {
		return nil, ErrNoAvailableEndpoints
	}

	return newClient(&cfg)
}

// NewCtxClient创建一个具有上下文但没有基础grpc的客户端
// 连接。这对于覆盖
// 服务接口实现且不需要连接管理的嵌入式案例非常有用。
func NewCtxClient(ctx context.Context, opts ...Option) *Client {
	cctx, cancel := context.WithCancel(ctx)
	c := &Client{ctx: cctx, cancel: cancel, lgMu: new(sync.RWMutex)}
	for _, opt := range opts {
		opt(c)
	}
	if c.lg == nil {
		c.lg = zap.NewNop()
	}
	return c
}

// 选项是一种函数类型，可以作为参数传递给NewCtxClient以配置客户端
type Option func(*Client)

// NewFromURL从URL创建新的etcdv3客户端。
func NewFromURL(url string) (*Client, error) {
	return New(Config{Endpoints: []string{url}})
}

// NewFromURLs从URL创建一个新的etcdv3客户端。
func NewFromURLs(urls []string) (*Client, error) {
	return New(Config{Endpoints: urls})
}

// WithZapLogger是一个新的CTXClient选项，它覆盖记录器
func WithZapLogger(lg *zap.Logger) Option {
	return func(c *Client) {
		c.lg = lg
	}
}

// WithLogger覆盖记录器。
// 
// 已弃用：请在clientv3中使用WithZapLogger或Logger字段。Config 
// 
// 不会更改grpcLogger，可以使用grpc_zap显式配置该记录器。replacegrploggerv2（..）方法
func (c *Client) WithLogger(lg *zap.Logger) *Client {
	c.lgMu.Lock()
	c.lg = lg
	c.lgMu.Unlock()
	return c
}

// GetLogger获取记录器。
// 注意：此方法仅供etcd客户端库内部使用，不应用作通用记录器。
func (c *Client) GetLogger() *zap.Logger {
	c.lgMu.RLock()
	l := c.lg
	c.lgMu.RUnlock()
	return l
}

// 关闭关闭客户端的etcd连接。
func (c *Client) Close() error {
	c.cancel()
	if c.Watcher != nil {
		c.Watcher.Close()
	}
	if c.Lease != nil {
		c.Lease.Close()
	}
	if c.conn != nil {
		return toErr(c.ctx, c.conn.Close())
	}
	return c.ctx.Err()
}

// Ctx是“带外”消息的上下文（例如，在取消另一上下文时发送
// “清理”消息）。它是
// 在客户端关闭（）时取消。
func (c *Client) Ctx() context.Context { return c.ctx }

// Endpoints列出了客户端的已注册端点。
func (c *Client) Endpoints() []string {
	// 复制切片；保护原始端点不被更改
	c.epMu.RLock()
	defer c.epMu.RUnlock()
	eps := make([]string, len(c.endpoints))
	copy(eps, c.endpoints)
	return eps
}

// SetEndpoints更新客户端的端点。
func (c *Client) SetEndpoints(eps ...string) {
	c.epMu.Lock()
	defer c.epMu.Unlock()
	c.endpoints = eps

	c.resolver.SetEndpoints(eps)
}

// Client.Sync （） 方法中会向集群请求当前的节点列表，并更新本地的缓存，
func (c *Client) Sync(ctx context.Context) error {
	mresp, err := c.MemberList(ctx)	// 通过 GRPC 请求集群的节点列表
	if err != nil {
		return err
	}
	var eps []string
	for _, m := range mresp.Members {	// 遍历集群返回的节点列表
		eps = append(eps, m.ClientURLs...)
	}
	c.SetEndpoints(eps...)	// 更新 Client 记录的 URL 集合
	return nil
}

// 启动的后台 goroutine
func (c *Client) autoSync() {
	// 检测是否开后了自动同步
	if c.cfg.AutoSyncInterval == time.Duration(0) {
		return
	}

	for {
		select {
		case <-c.ctx.Done():
			return
		case <-time.After(c.cfg.AutoSyncInterval):	// 间隔一段时间，进行一次同步
			ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
			// 调用 Sync （）方法进行同步
			err := c.Sync(ctx)
			cancel()
			if err != nil && err != c.ctx.Err() {
				c.lg.Info("Auto sync endpoints failed.", zap.Error(err))
			}
		}
	}
}

// dialSetupOpts在任何身份验证之前提供拨号选项。
func (c *Client) dialSetupOpts(creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) (opts []grpc.DialOption, err error) {
	if c.cfg.DialKeepAliveTime > 0 {
		params := keepalive.ClientParameters{
			Time:                c.cfg.DialKeepAliveTime,
			Timeout:             c.cfg.DialKeepAliveTimeout,
			PermitWithoutStream: c.cfg.PermitWithoutStream,
		}
		opts = append(opts, grpc.WithKeepaliveParams(params))
	}
	opts = append(opts, dopts...)

	if creds != nil {
		opts = append(opts, grpc.WithTransportCredentials(creds))
	} else {
		opts = append(opts, grpc.WithInsecure())
	}

	// 侦听器重试并退出。
	// TODO:替换所有客户端v3/重试。go with RetryPolicy:
	// https:
	rrBackoff := withBackoff(c.roundRobinQuorumBackoff(defaultBackoffWaitBetween, defaultBackoffJitterFraction))
	opts = append(opts,
		// 默认情况下禁用流重试，因为go grpc中间件/重试不支持客户端流。
		// 可以安全重试的流被单独启用。
		grpc.WithStreamInterceptor(c.streamClientInterceptor(withMax(0), rrBackoff)),
		grpc.WithUnaryInterceptor(c.unaryClientInterceptor(withMax(defaultUnaryMaxRetries), rrBackoff)),
	)

	return opts, nil
}

// 建立网络连接的
func (c *Client) Dial(ep string) (*grpc.ClientConn, error) {
	creds := c.credentialsForEndpoint(ep)

	// 使用特别创建的解析器，以保证只使用显式给定的
	// 端点。
	return c.dial(creds, grpc.WithResolvers(resolver.New(ep)))
}

func (c *Client) getToken(ctx context.Context) error {
	var err error // 返回失败时的最后一个错误

	if c.Username == "" || c.Password == "" {
		return nil
	}

	resp, err := c.Auth.Authenticate(ctx, c.Username, c.Password)
	if err != nil {
		if err == rpctypes.ErrAuthNotEnabled {
			return nil
		}
		return err
	}
	c.authTokenBundle.UpdateAuthToken(resp.Token)
	return nil
}

// dialWithBalancer拨号客户端当前的负载平衡解析程序组。所提供端点的主机
// 的方案决定用于客户端连接的所有端点的方案。
func (c *Client) dialWithBalancer(dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
	creds := c.credentialsForEndpoint(c.Endpoints()[0])
	opts := append(dopts, grpc.WithResolvers(c.resolver))
	return c.dial(creds, opts...)
}

// 拨号配置并拨号任何grpc平衡器目标。
func (c *Client) dial(creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
	opts, err := c.dialSetupOpts(creds, dopts...)
	if err != nil {
		return nil, fmt.Errorf("failed to configure dialer: %v", err)
	}
	if c.authTokenBundle != nil {
		opts = append(opts, grpc.WithPerRPCCredentials(c.authTokenBundle.PerRPCCredentials()))
	}

	opts = append(opts, c.cfg.DialOptions...)

	dctx := c.ctx
	if c.cfg.DialTimeout > 0 {
		var cancel context.CancelFunc
		dctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
		defer cancel() // TODO:这是否适用于grpc。拨号选项上未设置WithBlock（）？
	}
	target := fmt.Sprintf("%s:// /%p/%s”，resolver.Schema，c，authority（c.endpoints[0]）
	conn, err := grpc.DialContext(dctx, target, opts...)
	if err != nil {
		return nil, err
	}
	return conn, nil
}

func authority(endpoint string) string {
	spl := strings.SplitN(endpoint, ":// /“，2）
	if len(spl) < 2 {
		if strings.HasPrefix(endpoint, "unix:") {
			return endpoint[len("unix:"):]
		}
		if strings.HasPrefix(endpoint, "unixs:") {
			return endpoint[len("unixs:"):]
		}
		return endpoint
	}
	return spl[1]
}

func (c *Client) credentialsForEndpoint(ep string) grpccredentials.TransportCredentials {
	r := endpoint.RequiresCredentials(ep)
	switch r {
	case endpoint.CREDS_DROP:
		return nil
	case endpoint.CREDS_OPTIONAL:
		return c.creds
	case endpoint.CREDS_REQUIRE:
		if c.creds != nil {
			return c.creds
		}
		return credentials.NewBundle(credentials.Config{}).TransportCredentials()
	default:
		panic(fmt.Errorf("unsupported CredsRequirement: %v", r))
	}
}

// Client 实例的初始化过程是在 newClient（）函 数中完成的 ， 该方法首先会创建 Client 实例，
// 然后创建 grpc . ClientConn 实例，最后初始化 GRPC 服务的客户端。
func newClient(cfg *Config) (*Client, error) {
	// 检测配置信息
	if cfg == nil {
		cfg = &Config{}
	}
	var creds grpccredentials.TransportCredentials
	if cfg.TLS != nil {
		creds = credentials.NewBundle(credentials.Config{TLSConfig: cfg.TLS}).TransportCredentials()
	}

	// 使用临时框架客户端引导第一个连接
	baseCtx := context.TODO()
	if cfg.Context != nil {
		baseCtx = cfg.Context
	}

	ctx, cancel := context.WithCancel(baseCtx)
	client := &Client{	// 创建 Client 实例，注意，此时的 conn 字段还未初始化
		conn:     nil,
		cfg:      *cfg,
		creds:    creds,
		ctx:      ctx,
		cancel:   cancel,
		epMu:     new(sync.RWMutex),
		callOpts: defaultCallOpts,
		lgMu:     new(sync.RWMutex),
	}

	var err error
	if cfg.Logger != nil {
		client.lg = cfg.Logger
	} else if cfg.LogConfig != nil {
		client.lg, err = cfg.LogConfig.Build()
	} else {
		client.lg, err = CreateDefaultZapLogger()
	}
	if err != nil {
		return nil, err
	}

	if cfg.Username != "" && cfg.Password != "" {	// 记录配置信息中携带的用户名和密码
		client.Username = cfg.Username
		client.Password = cfg.Password
		client.authTokenBundle = credentials.NewBundle(credentials.Config{})
	}
	if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
		if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
			return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
		}
		callOpts := []grpc.CallOption{
			defaultWaitForReady,
			defaultMaxCallSendMsgSize,
			defaultMaxCallRecvMsgSize,
		}
		if cfg.MaxCallSendMsgSize > 0 {
			callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
		}
		if cfg.MaxCallRecvMsgSize > 0 {
			callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
		}
		client.callOpts = callOpts
	}

	client.resolver = resolver.New(cfg.Endpoints...)

	if len(cfg.Endpoints) < 1 {
		client.cancel()
		return nil, fmt.Errorf("at least one Endpoint is required in client config")
	}
	client.SetEndpoints(cfg.Endpoints...)

	// 使用提供的端点目标，以便对于https:
	// grpc将假定证书服务器名称为端点主机。
	conn, err := client.dialWithBalancer()
	if err != nil {
		client.cancel()
		client.resolver.Close()
		// TODO:类似'fmt'的错误。Errorf（拨号[%s]失败：%v，strings.Join（cfg.Endpoints，“；”），呃）`这对调试很有帮助。ABCFDG 
		return nil, err
	}
	client.conn = conn	// 初始化 Client.conn 字段
	// 初始化多个 gRPC 服务的客户将
	client.Cluster = NewCluster(client)
	client.KV = NewKV(client)
	client.Lease = NewLease(client)
	client.Watcher = NewWatcher(client)
	client.Auth = NewAuth(client)
	client.Maintenance = NewMaintenance(client)

	ctx, cancel = client.ctx, func() {}
	if client.cfg.DialTimeout > 0 {	// 阻塞等待连接建立完成
		ctx, cancel = context.WithTimeout(ctx, client.cfg.DialTimeout)
	}
	err = client.getToken(ctx)
	if err != nil {
		client.Close()
		cancel()
		return nil, err
	}
	cancel()

	if cfg.RejectOldCluster {
		if err := client.checkVersion(); err != nil {
			client.Close()
			return nil, err
		}
	}
	// 在这里启动一个后台 goroutine ，定时同步集群中各个节点暴露的 URL
	go client.autoSync()
	return client, nil
}

// 在每次退避之间根据仲裁重试RoundRobinQuorUnbackoff。
// 这是用于循环负载平衡器。
func (c *Client) roundRobinQuorumBackoff(waitBetween time.Duration, jitterFraction float64) backoffFunc {
	return func(attempt uint) time.Duration {
		// 在通过仲裁的每一轮循环之后，为我们在持续时间之间的等待退避
		n := uint(len(c.Endpoints()))
		quorum := (n/2 + 1)
		if attempt%quorum == 0 {
			c.lg.Debug("backoff", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum), zap.Duration("waitBetween", waitBetween), zap.Float64("jitterFraction", jitterFraction))
			return jitterUp(waitBetween, jitterFraction)
		}
		c.lg.Debug("backoff skipped", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum))
		return 0
	}
}

func (c *Client) checkVersion() (err error) {
	var wg sync.WaitGroup

	eps := c.Endpoints()
	errc := make(chan error, len(eps))
	ctx, cancel := context.WithCancel(c.ctx)
	if c.cfg.DialTimeout > 0 {	// 阻塞等待连接建立完成
		cancel()
		ctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
	}

	wg.Add(len(eps))
	for _, ep := range eps {
		// 如果集群是当前的，则任何端点都会提供最新版本
		go func(e string) {
			defer wg.Done()
			resp, rerr := c.Status(ctx, e)
			if rerr != nil {
				errc <- rerr
				return
			}
			vs := strings.Split(resp.Version, ".")
			maj, min := 0, 0
			if len(vs) >= 2 {
				var serr error
				if maj, serr = strconv.Atoi(vs[0]); serr != nil {
					errc <- serr
					return
				}
				if min, serr = strconv.Atoi(vs[1]); serr != nil {
					errc <- serr
					return
				}
			}
			if maj < 3 || (maj == 3 && min < 2) {
				rerr = ErrOldCluster
			}
			errc <- rerr
		}(ep)
	}
	// 等待成功
	for range eps {
		if err = <-errc; err == nil {
			break
		}
	}
	cancel()
	wg.Wait()
	return err
}

// ActiveConnection返回当前正在使用的连接
func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }

// 如果给定的错误和上下文指示没有向前
// 即使在重新连接后也无法取得进展，则ISHATER返回true。
func isHaltErr(ctx context.Context, err error) bool {
	if ctx != nil && ctx.Err() != nil {
		return true
	}
	if err == nil {
		return false
	}
	ev, _ := status.FromError(err)
	// 不可用的代码表示系统将立即返回。
	// （例如，无法连接，失去领导）
	// 将内部代码视为出现故障，使
	// 系统处于不一致状态，但重试可能会取得进展。
	// （例如，发送过程中失败，帧损坏）
	// TODO:grpc是否可能出现永久性内部错误？
	return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
}

// isUnavailableErr如果给定的错误是不可用错误，则返回true 
func isUnavailableErr(ctx context.Context, err error) bool {
	if ctx != nil && ctx.Err() != nil {
		return false
	}
	if err == nil {
		return false
	}
	ev, ok := status.FromError(err)
	if ok {
		// 不可用代码表示系统将立即返回。
		// （例如，无法连接，失去领导）
		return ev.Code() == codes.Unavailable
	}
	return false
}

func toErr(ctx context.Context, err error) error {
	if err == nil {
		return nil
	}
	err = rpctypes.Error(err)
	if _, ok := err.(rpctypes.EtcdError); ok {
		return err
	}
	if ev, ok := status.FromError(err); ok {
		code := ev.Code()
		switch code {
		case codes.DeadlineExceeded:
			fallthrough
		case codes.Canceled:
			if ctx.Err() != nil {
				err = ctx.Err()
			}
		}
	}
	return err
}

func canceledByCaller(stopCtx context.Context, err error) bool {
	if stopCtx.Err() == nil || err == nil {
		return false
	}

	return err == context.Canceled || err == context.DeadlineExceeded
}

// IsConnCanceled返回true，如果错误来自关闭的gRPC连接。
// 参考。https:
func IsConnCanceled(err error) bool {
	if err == nil {
		return false
	}

	// />=gRPC v1。23.x 
	s, ok := status.FromError(err)
	if ok {
		// 连接已取消或服务器已关闭连接
		return s.Code() == codes.Canceled || s.Message() == "transport is closing"
	}

	// />=gRPC v1。10.x 
	if err == context.Canceled {
		return true
	}

	// /<=gRPC v1。7.x返回“错误”。新建（“grpc:客户端连接正在关闭”）'
	return strings.Contains(err.Error(), "grpc: the client connection is closing")
}
